package com.cdy.client.netty;

import com.cdy.client.handler.HandlerManager;
import com.cdy.common.command.Command;
import com.cdy.common.command.RequestDownloadCommand;
import com.cdy.common.command.RequestUploadCommand;
import com.cdy.common.property.CommonProperty;
import com.cdy.common.response.Response;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * netty 客户端
 * <p>
 * Created by 陈东一
 * 2019/10/19 0019 16:09
 */
@Slf4j
public class NettyClient {

    private static final NioEventLoopGroup boss = new NioEventLoopGroup();
    public static final  AtomicLong idAdder = new AtomicLong();
    public static final ConcurrentHashMap<Long, CompletableFuture<Response>> responseFuture = new ConcurrentHashMap<>();
    public static final AttributeKey<NettyClient> NETTY_CLIENT = AttributeKey.valueOf("nettyClient");

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(boss::shutdownGracefully));
    }

    private final Bootstrap bootstrap = new Bootstrap();
    @Getter private final CommonProperty commonProperty;
    private Boolean isStart = false;
    @Getter private Boolean isClose = false;
    private Channel channel;
    private String ip;
    private Integer port;

    public NettyClient(CommonProperty commonProperty) {
        this.commonProperty = commonProperty;
    }

    public void connect(String ip, Integer port) throws InterruptedException {
        NettyClient client = this;
        this.ip = ip;
        this.port= port;
        if (isStart) {
            throw new RuntimeException("client is already started");
        }
        bootstrap.channel(NioSocketChannel.class)
                .group(boss.next())
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(0, Math.toIntExact(CommonProperty.BigSectionLength)))
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline()
//                                .addLast(new WriteTimeoutHandler(60))
                                .addLast(new ClientDecodeHandler())
                                .addLast(new ClientEncodeHandler())
                                .addLast(new ClientHandler(new HandlerManager()))
                                .addLast(new ClientErrorHandler())
                        ;

                    }
                });

        ChannelFuture future = bootstrap.connect(ip, port);
        future.addListener(f -> reconnect()); //启动时连接失败，重新连接
        channel = future.channel();
        channel.attr(NETTY_CLIENT).set(this);
        log.info("client channel id={}, {} ",channel.id(),channel.toString());
        isStart = true;
    }

    public void reconnect() throws InterruptedException {
        ChannelFuture future = bootstrap.connect(ip, port).sync();
        channel = future.channel();
        channel.attr(NETTY_CLIENT).set(this);
        log.info("client channel id={}, {} ",channel.id(),channel.toString());
    }

    public void close() {
        if (isClose) {
            throw new RuntimeException("client is already closed");
        }
        isClose = true;
        isStart = false;
        channel.close();
    }

    public Optional<?> send(Command command) {
        if (command instanceof RequestUploadCommand) {
            long requestId = idAdder.incrementAndGet();
            CompletableFuture<Response> future = new CompletableFuture<>();
            responseFuture.put(requestId, future);
            ((RequestUploadCommand) command).setRequestId(requestId);
            write(channel,command);
            return Optional.of(future);
        } else if (command instanceof RequestDownloadCommand) {
            long requestId = idAdder.incrementAndGet();
            CompletableFuture<Response> future = new CompletableFuture<>();
            responseFuture.put(requestId, future);
            ((RequestDownloadCommand) command).setRequestId(requestId);
            write(channel,command);
            return Optional.of(future);
        }else {
            write(channel,command);
            return Optional.empty();
        }
    }

    private void write(Channel channel, Command response){
        if (channel.isWritable()) {
            channel.writeAndFlush(response);
        } else {
            channel.eventLoop().schedule(()->write(channel, response), 1, TimeUnit.MINUTES);
        }
    }
}
